// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_ROOT_META_TABLET_CLIENT_HPP_
#define _XBASE_ROOT_META_TABLET_CLIENT_HPP_ "hepengfei"

#include <boost/asio.hpp>
#include <boost/lexical_cast.hpp>

#include <xsocket/bytebuffer.hpp>

#include <protocol.hpp>
#include <xbase_datagram_factory.hpp>
#include <constants.hpp>
#include <logger.hpp>

#include <protocol_packet.hpp>

#include "tablet_client.hpp"

namespace xbase
{
	class root_meta_tablet_client : public tablet_client
	{
	public:
		root_meta_tablet_client(const std::string &p_hostname,
					const uint16_t &p_port)
			: m_io_service()
			, m_socket(m_io_service) {
			boost::asio::ip::tcp::resolver l_resolver(m_io_service);
			boost::asio::ip::tcp::resolver::query l_query(boost::asio::ip::tcp::v4(),
								      p_hostname,
								      boost::lexical_cast<std::string>(p_port));
			boost::asio::ip::tcp::resolver::iterator l_endpoint_iterator = l_resolver.resolve(l_query);
			
			boost::system::error_code l_error;
			m_socket.connect(*l_endpoint_iterator, l_error);
			if (l_error)
			{
				m_socket.close();
				//throw boost::system::system_error(l_error);
			}
		}

		virtual ~root_meta_tablet_client() {
			m_socket.close();
		}

		bool is_open() const {
			return m_socket.is_open();
		}

		// 只更新子表信息
		bool update_tablet(const std::string &p_tablet_name,
				   const protocol::tablet_info &p_tablet,
				   const std::string &p_log_msg) {
			return update_tablet(p_tablet_name,
					     p_log_msg,
					     &p_tablet,
					     NULL,
					     NULL,
					     NULL,
					     NULL);
		}

		bool update_tablet(const std::string &p_tablet_name,
				   const protocol::tablet_info &p_tablet,
				   const protocol::server_address &p_address,
				   uint64_t p_start_code,
				   const std::string &p_log_msg) {
			return update_tablet(p_tablet_name,
					     p_log_msg,
					     &p_tablet,
					     &p_address,
					     &p_start_code,
					     NULL,
					     NULL);
		}

		bool update_tablet(const std::string &p_tablet_name,
				   const protocol::tablet_info &p_tablet,
				   const protocol::server_address &p_address,
				   uint64_t p_start_code,
				   const protocol::tablet_info &p_split_a,
				   const protocol::tablet_info &p_split_b,
				   const std::string &p_log_msg) {
			return update_tablet(p_tablet_name,
					     p_log_msg,
					     &p_tablet,
					     &p_address,
					     &p_start_code,
					     &p_split_a,
					     &p_split_b);
		}

		bool update_tablet(const std::string &p_tablet_name,
				   const std::string &p_log_msg,
				   const protocol::tablet_info *p_tablet_info,
				   const protocol::server_address *p_address,
				   uint64_t *p_start_code,
				   const protocol::tablet_info *p_split_a,
				   const protocol::tablet_info *p_split_b) {
			typedef protocol::packet_type<protocol::WRITE_REQ>::type write_req;
			typedef protocol::packet_type<protocol::WRITE_RESP>::type write_resp;

			write_req req;
			write_req::value_type &body = req.body();
			protocol::row_mutation_data_ptr ptr(new protocol::row_mutation_data(p_tablet_name));
			
			body.m_tablet_name = ROOT_META_TABLET_NAME;
			body.m_timestamp = cur_micro_seconds();
			body.m_mutation_data = ptr;

			const std::size_t max_packet_size = 1024 * 1024;
			const std::size_t max_value_size = 1024 * 512; // max value size in root meta tablet
			simple_buffer buffer;
			buffer.prepare(max_packet_size);

			std::size_t record_count = 0;

			if(p_tablet_info != NULL)
			{
				protocol::oarchive out(buffer.prepare(max_value_size));
				LOG_ENV(bool ok = )
					out.save(*p_tablet_info);
				LOG_IF((!ok), FATAL,
				       "encode tablet info failed, tablet info = "
				       << p_tablet_info->to_string());
				assert(ok);
				const std::string value(out.used_buffer(), out.used_buffer_size());
				ptr->add(".DEFAULT_GROUP", "info:tabletInfo", value, NO_TIMESTAMP);
				++record_count;
			}

			if(p_address != NULL)
			{
				protocol::oarchive out(buffer.prepare(max_value_size));
				LOG_ENV(bool ok = )
					out.save(*p_address);
				LOG_IF((!ok), FATAL,
				       "encode server address failed, address = "
				       << p_address->to_string());
				assert(ok);
				const std::string value(out.used_buffer(), out.used_buffer_size());
				ptr->add(".DEFAULT_GROUP", "info:serverAddress", value, NO_TIMESTAMP);
				++record_count;
			}

			if(p_start_code != NULL)
			{
				protocol::oarchive out(buffer.prepare(max_value_size));
				LOG_ENV(bool ok = )
					out.save(*p_start_code);
				LOG_IF((!ok), FATAL,
				       "encode start code failed, start code = "
				       << *p_start_code);
				assert(ok);
				const std::string value(out.used_buffer(), out.used_buffer_size());
				ptr->add(".DEFAULT_GROUP", "info:serverStartCode", value, NO_TIMESTAMP);
				++record_count;
			}

			if(p_split_a != NULL)
			{
				protocol::oarchive out(buffer.prepare(max_value_size));
				const std::string tablet_name = p_split_a->tablet_name();
				LOG_ENV(bool ok = )
					out.save(tablet_name);
				LOG_IF((!ok), FATAL,
				       "encode split_a tablet_name failed, split_a = "
				       << p_split_a->to_string() << ", tablet_name = "
				       << tablet_name);
				assert(ok);
				const std::string value(out.used_buffer(), out.used_buffer_size());
				ptr->add(".DEFAULT_GROUP", "info:splitA", value, NO_TIMESTAMP);
				++record_count;
			}

			if(p_split_b != NULL)
			{
				protocol::oarchive out(buffer.prepare(max_value_size));
				const std::string tablet_name = p_split_b->tablet_name();
				LOG_ENV(bool ok = )
					out.save(tablet_name);
				LOG_IF((!ok), FATAL,
				       "encode split_b tablet_name failed, split_b = "
				       << p_split_b->to_string() << ", tablet_name = "
				       << tablet_name);
				assert(ok);
				const std::string value(out.used_buffer(), out.used_buffer_size());
				ptr->add(".DEFAULT_GROUP", "info:splitB", value, NO_TIMESTAMP);
				++record_count;
			}

			if(true)
			{
				ptr->add("debug", "logs:message", p_log_msg, NO_TIMESTAMP);
				++record_count;
			}

			// encode
			protocol::oarchive out(buffer.prepare(max_packet_size));
			LOG_ENV(bool ok = )
				out.save(req);
			LOG_IF((!ok), FATAL,
			       "encode write request failed, req = "
			       << req.to_string());
			assert(ok);
			const std::size_t req_size = out.used_buffer_size();
			buffer.commit(req_size);

			// send and receive
			boost::system::error_code ec;
			std::size_t len = boost::asio::write(m_socket,
							     buffer.data(),
							     boost::asio::transfer_all(),
							     ec);
			if(ec) {
				LOG(ERROR, "update_tablet failed: write data failed"
					<< ec << " - " << ec.message());
				return false; // error
			}
			buffer.clear();

			len = boost::asio::read(m_socket,
						buffer.prepare(max_packet_size),
						boost::asio::transfer_at_least(protocol::packet::HEADER_SIZE),
						ec);
			if(ec)
			{
				LOG(ERROR, "update_tablet failed: read response failed"
					<< ec << " - " << ec.message());
				return false;
			}
			buffer.commit(len);
			const boost::asio::const_buffer tmp_data = buffer.data();
			const char *start = boost::asio::buffer_cast<const char*>(tmp_data);
			const char *end = start + boost::asio::buffer_size(tmp_data);
			std::size_t bytes_to_read = 0;
			std::size_t id = 0;
			protocol::validate(start, &end,
					   &bytes_to_read, &id, ec);
			if((ec == error::packet_incomplete) && (bytes_to_read != 0))
			{
				len = boost::asio::read(m_socket,
							buffer.prepare(buffer.left_space()),
							boost::asio::transfer_at_least(bytes_to_read),
							ec);
				buffer.commit(len);
			}
			if(ec)
			{
				LOG(ERROR, "update_tablet failed: validate or read response data failed: "
				    << ec << " - " << ec.message());
				return false;
			}

			write_resp resp;
			protocol::iarchive in(buffer.data());
			if(! in.load(&resp))
			{
				LOG(ERROR, "update_tablet error: decode response failed");
				return false;
			}

			if(resp.body().error_code() != error::ok)
			{
				LOG(ERROR, "update_tablet error: response = "
				    << resp.to_string());
				return false; // error
			}

			// success
			LOG(DEBUG, "update_tablet succeeded, " << record_count
			    << " record written");
			return true;
		}

		bool get_tablet(protocol::tablet_info &tablet_info,
				protocol::server_address &address,
				uint64_t &startcode,
				const std::string &tablet_name) {
			typedef protocol::packet_type<protocol::READ_REQ>::type read_req;
			typedef protocol::packet_type<protocol::READ_RESP>::type read_resp;

			read_req req;
			req.body().m_tablet_name = ROOT_META_TABLET_NAME;
			req.body().m_key = protocol::key(tablet_name, "info:*", NO_TIMESTAMP);

			const std::size_t max_packet_size = 1024 * 1024;
			simple_buffer buffer;

			// encode
			protocol::oarchive out(buffer.prepare(max_packet_size));
			if(! out.save(req))
			{
				LOG(ERROR, "encode read request failed, request = "
				    << req.to_string());
				assert(false);
			}
			buffer.commit(out.used_buffer_size());

			// send req
			boost::system::error_code ec;
			std::size_t len = boost::asio::write(m_socket,
							     buffer.data(),
							     boost::asio::transfer_all(),
							     ec);
			if(ec)
			{
				LOG(ERROR, "send read request failed, ec = "
				    << ec << " - " << ec.message());
				return false;
			}
			buffer.clear();

			// receive resp
			len = boost::asio::read(m_socket,
						buffer.prepare(max_packet_size),
						boost::asio::transfer_at_least(protocol::packet::HEADER_SIZE),
						ec);
			if(ec)
			{
				LOG(ERROR, "read response failed, ec = "
				    << ec << " - " << ec.message());
				return false;
			}
			buffer.commit(len);
			const boost::asio::const_buffer tmp_data = buffer.data();
			const char *start = boost::asio::buffer_cast<const char*>(tmp_data);
			const char *end = start + boost::asio::buffer_size(tmp_data);
			std::size_t bytes_to_read = 0;
			std::size_t id = 0;
			protocol::validate(start, &end,
					   &bytes_to_read, &id, ec);
			if((ec == error::packet_incomplete) && (bytes_to_read != 0))
			{
				len = boost::asio::read(m_socket,
							buffer.prepare(buffer.left_space()),
							boost::asio::transfer_at_least(bytes_to_read),
							ec);
				buffer.commit(len);
			}
			if(ec)
			{
				LOG(ERROR, "validate or read response data failed: "
				    << ec << " - " << ec.message());
				return false;
			}

			// decode resp
			read_resp resp;
			protocol::iarchive in(buffer.data());
			if(! in.load(&resp))
			{
				LOG(ERROR, "decode read response failed");
				return false;
			}

			if(resp.body().error_code() != error::ok)
			{
				LOG(ERROR, "read error, response = "
				    << resp.to_string());
				return false;
			}

			// get value
			LOG_ENV(std::size_t readed_count = 0);
			for(std::size_t i = 0; i < resp.body().m_records.size(); ++i)
			{
#ifndef NDEBUG
				// 检查行名是否正确
				if(resp.body().m_records[i].m_key.m_row_name != tablet_name)
				{
					LOG(ERROR, "root_meta_tablet_client::get_tablet row name "
					    << resp.body().m_records[i].m_key.m_row_name
					    << " does not match tablet name "
					    << tablet_name);
					return false;
				}
#endif	// NDEBUG
				if(resp.body().m_records[i].m_key.m_column_name == "info:serverAddress")
				{
					protocol::iarchive in(resp.body().m_records[i].m_value.data(),
							      resp.body().m_records[i].m_value.size());
					LOG_ENV(bool ok = )
						in.load(&address);
					LOG_IF_ELSE((!ok),
						    ERROR, "decode info:serverAddress failed",
						    TRACE, "info:serverAddress readed");
					LOG_ENV(if(ok)++readed_count);
				}
				else if(resp.body().m_records[i].m_key.m_column_name == "info:tabletInfo")
				{
					protocol::iarchive in(resp.body().m_records[i].m_value.data(),
							      resp.body().m_records[i].m_value.size());
					LOG_ENV(bool ok = )
						in.load(&tablet_info);
					LOG_IF_ELSE((!ok),
						    ERROR, "decode info:tabletInfo failed",
						    TRACE, "info:tabletInfo readed");
					LOG_ENV(if(ok)++readed_count);
				}
				else if(resp.body().m_records[i].m_key.m_column_name == "info:serverStartCode")
				{
					protocol::iarchive in(resp.body().m_records[i].m_value.data(),
							      resp.body().m_records[i].m_value.size());
					LOG_ENV(bool ok = )
						in.load(&startcode);
					LOG_IF_ELSE((!ok),
						    ERROR, "decode info:serverStartCode failed",
						    TRACE, "info:serverStartCode readed");
					LOG_ENV(if(ok)++readed_count);
				}
				else
				{
					LOG(TRACE, "skip column " << resp.body().m_records[i].m_key.m_column_name);
				}
			}
			LOG(TRACE, readed_count << " records readed");
			return true;
		}

	protected:
		size_t write(const void *p_buffer, size_t p_len) {
			boost::system::error_code l_error;
			size_t l_len = boost::asio::write(m_socket,
							  boost::asio::buffer(p_buffer, p_len),
							  boost::asio::transfer_all(),
							  l_error);

			if (l_error == boost::asio::error::eof)
				;
			else
				;

			return l_len;
		}

		size_t read_some(void *p_buffer, size_t p_len) {
			boost::system::error_code l_error;
			size_t l_len = m_socket.read_some(boost::asio::buffer(p_buffer, p_len), l_error);

			if (l_error == boost::asio::error::eof)
				;
			else
				;

			return l_len;
		}

	private:
		boost::asio::io_service m_io_service;
		boost::asio::ip::tcp::socket m_socket;
		
	};
} // namespace xbase

#endif	// _XBASE_ROOT_META_TABLET_CLIENT_HPP_
